/********************************************************/
/*	Copyright (C) 2012 Gong Li Bin			*/
/*	Project:	MapReduce			*/
/*	Author:		gong_libin			*/
/*	Date:		2016_06_01			*/
/*	File:		MrProcess.cpp			*/
/********************************************************/

#include "MrProcess.h"

CMrUdp g_CUdp;
CMrSet g_CMap;
CMrSet g_CRed;
CMrSet g_CHub;
CMrSet g_CPoll;
MRJBPACKET_S g_stRed;
bool g_bDebug = false;

int main(int argc, char* argv[])
{
	int iCount = 0;
	int iIndex = 0;
	int iReturn = 0;
	int iThread = 0;
	CMrThread CAdmin;
	CMrThread CThread;
	CMrThread* pCThread = NULL;
	MRPRPACKET_S* pstPacket = NULL;
	struct sockaddr_in* pstAdp = NULL;
	char szPort[MR_VALUE] = { 0 };
	char szConf[MR_VALUE] = "MrProcess.conf";

	struct option stOptions[] = {
		{"conf", 1, 0, 'c'},
		{"thread", 1, 0, 't'},
		{"debug", 0, 0, 'd'},
		{"help", 0, 0, 'h'},
		{"version", 0, 0, 'v'},
		{0, 0, 0, 0},
	};

	while (true) {
		iReturn = getopt_long(argc, argv, "c:t:dhv", stOptions, &iIndex);
		if (iReturn < 0) {
			break;
		}
		switch (iReturn) {
			case 'c':
				strncpy(szConf, optarg, MR_VALUE);
				break;
			case 't':
				iThread = atoi(optarg);
				break;
			case 'd':
				g_bDebug = true;
				break;
			case 'h':
				MrJbHelp(argv[0], MR_PR_VERSION);
				return MR_SUCCESS;
			case 'v':
				MR_PRINT("Copyright (C) 2012 Gong Li Bin Technology Inc.\n");
				MR_PRINT("%s Build on %s %s.\n", MR_PR_VERSION, __DATE__, __TIME__);
				return MR_SUCCESS;
			default:
				return MR_SUCCESS;
		}
	}

	signal(SIGSEGV, MrJbQuit);
	signal(SIGTERM, MrJbQuit);

	/* config */
	if (true == g_bDebug) {
		MR_DEBUG("Initialize config from %s\n", szConf);
	}
	if (MR_SUCCESS == MrProcessConfig(szConf, szPort)) {
		if (true == g_bDebug) {
			MR_DEBUG("Port: %s\n", szPort);
		}
	}
	else {
		return MR_FAILURE;
	}

	/* fork */
	if (true != g_bDebug) {
		MrFork();
	}

	/* admin */
	if (true == g_bDebug) {
		MR_DEBUG("Create admin thread\n");
	}
	CAdmin.MrThreadAttrInit();
	CAdmin.MrThreadSetStartRoutine(MrJbAdminCore);
	if (MR_FAILURE == CAdmin.MrThreadCreate(NULL)) {
		return MR_FAILURE;
	}

	/* process */
	if (true == g_bDebug) {
		MR_DEBUG("MrProcess will listen...\n");
	}
	srand((UINT)time(NULL));
	g_CUdp.MrNetLimit(MR_SOCK_MAX);
	if (MR_SUCCESS == g_CUdp.MrNetInitRecv(NULL, szPort)) {
		if (0 == iThread) {
			/* dynamic */
			while (true) {
				if (NULL != (pstPacket = (MRPRPACKET_S*)MrMalloc(sizeof(MRPRPACKET_S)))) {
					if ((pstPacket->MR_PR_JOBLEN = g_CUdp.MrNetRecv(pstPacket->MR_PR_JOBPKT, MR_BIGPACKET)) > 0) {
						pstAdp = (struct sockaddr_in*)g_CUdp.MrUdpGetAddr();
						sprintf(pstPacket->m_unAddr.m_szUrl, "%s:%d", inet_ntoa(pstAdp->sin_addr), ntohs(pstAdp->sin_port));

						CThread.MrThreadAttrInit();
						CThread.MrThreadAttrSetDetach();
						CThread.MrThreadSetStartRoutine(MrProcessThread);
						CThread.MrThreadCreate((void*)pstPacket);
					}
					else {
						MR_ERROR("Failed to MrNetRecv\n");
					}
				}
				else {
					MR_ERROR("Failed to MrMalloc\n");
				}
			}
		}
		else {
			/* statics */
			pCThread = new CMrThread[iThread];
			for (iCount = 0; iCount < iThread; iCount ++) {
				(pCThread + iCount)->MrThreadAttrInit();
				(pCThread + iCount)->MrThreadSetStartRoutine(MrProcessThread);
				(pCThread + iCount)->MrThreadCreate(NULL);
			}
			for (iCount = 0; iCount < iThread; iCount ++) {
				(pCThread + iCount)->MrThreadJoin(NULL);
			}
		}
		g_CUdp.MrNetClose();
	}
	else {
		MR_ERROR("Failed to MrNetInitRecv\n");
	}

	return MR_SUCCESS;
}

int MrProcessConfig(char* pszConf, char* pszPort)
{
	CMrDb CDb;
	MRCONFIG_S stConf;
	USHORT usCount = 0;
	int iReturn = MR_SUCCESS;

	memset(&stConf, '\0', sizeof(MRCONFIG_S));
	if (MR_SUCCESS == MrConfig(pszConf, &stConf)) {
		if (MR_SUCCESS == CDb.MrDbConnect(stConf.m_szHost, stConf.m_szPort)) {
			/* port */
			iReturn = MrJbPortInit(CDb, MR_PR_PORT, &stConf, pszPort);
			/* poll */
			if (MR_SUCCESS == iReturn) {
				if (MR_SUCCESS == (iReturn = MrJbDevInit(CDb, MR_PR_POLL, &stConf, &g_CPoll, MR_JB_POLL))) {
					if (true == g_bDebug) {
						MR_DEBUG("Poll:");
						for (usCount = 0; usCount < g_CPoll.MrSetGetCount(); usCount ++) {
							MR_PRINT(" %d,", ((MRJBPOLL_S*)g_CPoll.MrSetIndexGet(usCount))->m_stHost.m_uiID);
						}
						MR_PRINT("\n");
					}
				}
			}
			/* map */
			if (MR_SUCCESS == iReturn) {
				if (MR_SUCCESS == (iReturn = MrJbDevInit(CDb, MR_PR_MAP, &stConf, &g_CMap, MR_JB_HOST))) {
					if (true == g_bDebug) {
						MR_DEBUG("Map:");
						for (usCount = 0; usCount < g_CMap.MrSetGetCount(); usCount ++) {
							MR_PRINT(" %d,", ((MRJBHOST_S*)g_CMap.MrSetIndexGet(usCount))->m_uiID);
						}
						MR_PRINT("\n");
					}
				}
			}
			/* red */
			if (MR_SUCCESS == iReturn) {
				if (MR_SUCCESS == (iReturn = MrJbDevInit(CDb, MR_PR_RED, &stConf, &g_CRed, MR_JB_HOST))) {
					if (true == g_bDebug) {
						MR_DEBUG("Red:");
						for (usCount = 0; usCount < g_CRed.MrSetGetCount(); usCount ++) {
							MR_PRINT(" %d,", ((MRJBHOST_S*)g_CRed.MrSetIndexGet(usCount))->m_uiID);
						}
						MR_PRINT("\n");
					}
					MrJbDevReqSet(&g_stRed, g_CRed);
				}
			}
			/* hub */
			if (MR_SUCCESS == iReturn) {
				if (MR_SUCCESS == (iReturn = MrProcessHubInit(CDb, MR_PR_HUB, &stConf, &g_CHub))) {
					if (true == g_bDebug) {
						MR_DEBUG("Hub:");
						for (usCount = 0; usCount < g_CHub.MrSetGetCount(); usCount ++) {
							MR_PRINT(" %s,", ((MRJBHOST_S*)g_CHub.MrSetIndexGet(usCount))->m_unAddr.m_szUrl);
						}
						MR_PRINT("\n");
					}
				}
			}
			CDb.MrDbClose();
		}
		else {
			MR_ERROR("Failed to MrDbConnect\n");
			iReturn = MR_FAILURE;
		}
	}
	else {
		MR_ERROR("Failed to MrConfig\n");
		iReturn = MR_FAILURE;
	}

	return iReturn;
}

int MrProcessHubInit(CMrDb& rCDb, char* pszSql, MRCONFIG_S* pstConf, CMrSet* pCSet)
{
	MRJBHOST_S stHub;
	UINT uiCount = 0;
	MRPROKV_S* pstKv = NULL;
	int iReturn = MR_SUCCESS;
	char szSql[MR_PACKET] = { 0 };
	char szValue[MR_VALUE] = { 0 };

	if (NULL != (pstKv = MrProKvMalloc(MR_BUFFER, MR_PACKET))) {
		sprintf(szSql, pszSql, pstConf->m_szDbase, pstConf->m_szTable);
		if (rCDb.MrDbSend(MR_DEF_SELT, szSql) > 0) {
			if (MR_SUCCESS == rCDb.MrDbRecv(pstKv)) {
				if (MR_SUCCESS == pCSet->MrSetInit(MR_JB_DEVICE, sizeof(MRJBHOST_S))) {
					while ('\0' != (pstKv + uiCount)->m_ucKey) {
						memset(&stHub, '\0', sizeof(MRJBHOST_S));
						/* ID */
						memcpy(szValue, (pstKv + uiCount)->m_puszValue, (pstKv + uiCount)->m_iLength);
						szValue[(pstKv + uiCount)->m_iLength] = '\0';
						stHub.m_uiID = atoi(szValue);
						uiCount += 1;
						/* Addr */
						memcpy(stHub.m_unAddr.m_szUrl, (pstKv + uiCount)->m_puszValue, (pstKv + uiCount)->m_iLength);
						stHub.m_unAddr.m_szUrl[(pstKv + uiCount)->m_iLength] = ':';
						uiCount += 1;
						/* port */
						memcpy(szValue, (pstKv + uiCount)->m_puszValue, (pstKv + uiCount)->m_iLength);
						szValue[(pstKv + uiCount)->m_iLength] = '\0';
						strcat(stHub.m_unAddr.m_szUrl, szValue);
						uiCount += 1;
						/* insert */
						pCSet->MrSetIndexPut(pCSet->MrSetGetCount(), &stHub);
					}
					pCSet->MrSetCompare(MrJbCompHost);
					pCSet->MrSetSort();
				}
				else {
					MR_ERROR("Failed to MrSetInit\n");
					iReturn = MR_FAILURE;
				}
			}
			else {
				MR_ERROR("Failed to MrDbRecv\n");
				iReturn = MR_FAILURE;
			}
		}
		else {
			MR_ERROR("Failed to MrDbSend\n");
			iReturn = MR_FAILURE;
		}

		MrProKvFree(&pstKv, MR_BUFFER);
	}
	else {
		MR_ERROR("Failed to MrProKvMalloc\n");
		iReturn = MR_FAILURE;
	}

	return iReturn;
}

void* MrProcessThread(void* pPara)
{
	if (NULL != pPara) {
		MrProcessDynamic(pPara);
	}
	else {
		MrProcessStatics();
	}

	pthread_exit(NULL);
}

void MrProcessStatics()
{
	CMrUdp CUdp;
	MRPRPACKET_S stPacket;
	struct sockaddr_in* pstAdp = NULL;

	CUdp.MrNetSetSocket(g_CUdp.MrNetGetSocket());
	while (true) {
		memset(&stPacket, '\0', sizeof(MRJBPACKET_S));
		if ((stPacket.MR_PR_JOBLEN = CUdp.MrNetRecv(stPacket.MR_PR_JOBPKT, MR_BIGPACKET)) > 0) {
			pstAdp = (struct sockaddr_in*)CUdp.MrUdpGetAddr();
			sprintf(stPacket.m_unAddr.m_szUrl, "%s:%d", inet_ntoa(pstAdp->sin_addr), ntohs(pstAdp->sin_port));
			MrProcessThreadCore(&stPacket);
		}
	}

	return;
}

void MrProcessDynamic(void* pPara)
{
	MrProcessThreadCore((MRPRPACKET_S*)pPara);
	MrFree(&pPara);

	return;
}

void MrProcessThreadCore(MRPRPACKET_S* pstPacket)
{
	CMrPro CPro;
	CMrTime CTime;
	int iResult = 0;
	MRPRJOB_S stMap;
	MRPRJOB_S stRed;
	UCHAR ucAttr = MR_PRO_0;
	MRPRJOB_S* pstJob = NULL;
	UINT uiRed[MR_PR_HOST] = { 0 };
	UCHAR uszResult[MR_BIGPACKET] = { 0 };

	if (true == g_bDebug) {
		MR_PRINT("\n");
		CTime.MrTimeOn();
		MR_DEBUG("Thread[%lu] start\n", pthread_self());
	}

	/* adp */
	memset(&stMap, '\0', sizeof(MRPRJOB_S));
	stMap.m_pszAdp = pstPacket->m_unAddr.m_szUrl;

	/* head */
	CPro.MrProReset();
	CPro.MrProParse(pstPacket->MR_PR_JOBPKT, pstPacket->MR_PR_JOBLEN);
	if (true == g_bDebug) {
		MR_DEBUG("<Request>:\n");
		CPro.MrProDisplay();
	}

	/* resp */
	ucAttr = CPro.MrProGetAttr();
	if (MR_PRO_0 != (ucAttr & MR_MR_ACC)) {
		MrJbResponse(pstPacket->m_unAddr.m_szUrl);
	}

	/* body */
	if (CPro.MrProGetLength() > 0) {
		/* 1. map init */
		if (true == g_bDebug) {
			MR_DEBUG("<Map create>\n");
		}
		if (MR_SUCCESS != MrProcessMapCreate(CPro, &stMap)) {
			CPro.MrProPutResult(MR_MR_MRC);
			goto MrError;
		}

		/* 2. map send */
		if (true == g_bDebug) {
			MR_DEBUG("<Map send>\n");
		}
		if (MR_SUCCESS != MrProcessJobSend(&stMap, g_CMap)) {
			CPro.MrProPutResult(MR_MR_MRS);
			goto MrError;
		}

		if (stMap.MR_PR_MRLEN > 0) {
			/* 3. map recv */
			if (true == g_bDebug) {
				MR_DEBUG("<Map recv>\n");
			}
			if (MR_SUCCESS != MrProcessJobRecv(&stMap)) {
				CPro.MrProPutResult(MR_MR_MRR);
				goto MrError;
			}

			if (MR_SUCCESS == MrProcessJobCheck(&stMap)) {
				if (MR_PRO_0 == (ucAttr & MR_MR_SUM)) {
					/* 4. red init */
					if (true == g_bDebug) {
						MR_DEBUG("<Red create>\n");
					}
					pstJob = &stRed;
					memset(&stRed, '\0', sizeof(MRPRJOB_S));
					if (MR_SUCCESS != MrProcessRedCreate(&stMap, &stRed)) {
						CPro.MrProPutResult(MR_MR_ERC);
						goto MrError;
					}

					/* 5. red send */
					if (true == g_bDebug) {
						MR_DEBUG("<Red send>\n");
					}
					if (MR_SUCCESS != MrProcessJobSend(&stRed, g_CRed)) {
						CPro.MrProPutResult(MR_MR_ERS);
						goto MrError;
					}

					/* 6. red recv */
					if (true == g_bDebug) {
						MR_DEBUG("<Red recv>\n");
					}
					if (MR_SUCCESS != MrProcessJobRecv(&stRed)) {
						CPro.MrProPutResult(MR_MR_ERR);
						goto MrError;
					}
				}
				else {
					pstJob = &stMap;
				}

				/* 7. res init */
				if (true == g_bDebug) {
					MR_DEBUG("<Res create>\n");
				}
				if (MR_SUCCESS != MrProcessResCreate(pstJob, uszResult, &iResult)) {
					CPro.MrProPutResult(MR_MR_EVC);
					goto MrError;
				}

				/* 8. res track */
				if (true == g_bDebug) {
					MR_DEBUG("<Res track>\n");
				}
				MrProcessRedInit(pstJob, uiRed);
				if (MR_SUCCESS != MrProcessResTrack(uszResult, iResult, uiRed)) {
					CPro.MrProPutResult(MR_MR_EVS);
					goto MrError;
				}
			}
			else {
				CPro.MrProPutResult(MR_DEF_NULL);
				goto MrError;
			}
		}
	}

MrError:
	if (MR_DEF_SUCC != CPro.MrProGetResult()) {
		if (true == g_bDebug) {
			MR_DEBUG("MrError!\n");
		}
		MrProcessError(stMap.m_pszAdp, CPro);
	}

	if (true == g_bDebug) {
		MR_DEBUG("Thread[%lu] end Use time ", pthread_self());
		CTime.MrTimeOff();
		CTime.MrTimeDisplay();
	}

	return;
}

int MrProcessMapCreate(CMrPro& CPro, MRPRJOB_S* pstMap)
{
	CMrUdp CUdp;
	USHORT usKv = 0;
	UCHAR ucMap = 0;
	UCHAR ucHost = 0;
	MRPROKV_S stOut;
	MRJBPACKET_S stPacket;
	int iReturn = MR_SUCCESS;
	char szValue[MR_VALUE] = { 0 };
	MRPROKV_S stIn[MR_JB_PAIR];
	MRPROKV_S* pstLcl[MR_JB_PAIR] = { NULL };

	memset(stIn, '\0', sizeof(MRPROKV_S) * MR_JB_PAIR);
	MrProKvInput(stIn, CPro.MrProGetTotal(), CPro.MrProGetBody(), CPro.MrProGetLength());

	if (true == g_bDebug) {
		MR_DEBUG("MrProKvInput:");
	}
	while (usKv < CPro.MrProGetTotal()) {
		switch (stIn[usKv].m_ucKey) {
			case MR_MR_ADP:
				if (true == g_bDebug) {
					MR_PRINT(" MR_MR_ADP,");
				}
				pstMap->m_pszAdp[stIn[usKv].m_iLength] = '\0';
				memcpy(pstMap->m_pszAdp, stIn[usKv].m_puszValue, stIn[usKv].m_iLength);
				break;
			case MR_MR_MR:
				if (true == g_bDebug) {
					MR_PRINT(" MR_MR_MR,");
				}
				pstMap->m_stGlobal.m_iLength = stIn[usKv].m_iLength;
				memcpy(pstMap->m_stGlobal.m_uszPacket, stIn[usKv].m_puszValue, stIn[usKv].m_iLength);
				break;
			case MR_MR_MAP:
				if (true == g_bDebug) {
					MR_PRINT(" MR_MR_MAP,");
				}
				szValue[stIn[usKv].m_iLength] = '\0';
				memcpy(szValue, stIn[usKv].m_puszValue, stIn[usKv].m_iLength);
				pstMap->m_stTask[pstMap->m_ucTotal].m_uiHost[ucHost ++] = atoi(szValue);
				break;
			case MR_MR_LCL:
				if (true == g_bDebug) {
					MR_PRINT(" MR_MR_LCL,");
				}
				ucHost = 0;
				pstLcl[pstMap->m_ucTotal ++] = &stIn[usKv];
				break;
			default:
				if (true == g_bDebug) {
					MR_PRINT(" MR_MR_NLL,");
				}
				break;
		}
		usKv += 1;
	}
	if (true == g_bDebug) {
		MR_PRINT("\n");
	}

	if (true == g_bDebug) {
		MR_DEBUG("MrAdapter: %s\n", pstMap->m_pszAdp);
	}

	if (MR_SUCCESS == CUdp.MrNetOpen(MR_SOCK_DGRAM)) {
		CUdp.MrNetSetTimeOut(0, MR_USEC);
		MrProcessHubSort(CUdp, pstMap->m_CHub);
		while (ucMap < pstMap->m_ucTotal) {
			if (true == g_bDebug) {
				MR_DEBUG("MapCreate map[%d]:", ucMap);
			}

			MR_PR_CURMLL = MR_PRO_HEAD;
			CPro.MrProPutTotal(MR_PRO_0);

			/* adp */
			if (0 == pstMap->MR_PR_MRLEN) {
				if (true == g_bDebug) {
					MR_PRINT(" MR_MR_ADP,");
				}
				MR_PR_TOTALPP;
				stOut.m_ucKey = MR_MR_ADP;
				stOut.m_iLength = strlen(pstMap->m_pszAdp);
				stOut.m_puszValue = (UCHAR*)pstMap->m_pszAdp;
				MR_PR_CURMLL += MrProKvOutput(&stOut, MR_PRO_1, MR_PR_CURMLP + MR_PR_CURMLL, MR_BIGPACKET - MR_PR_CURMLL);
			}

			/* hub */
			if (pstMap->m_CHub.MrSetGetCount() > 0) {
				usKv = 0;
				stOut.m_ucKey = MR_MR_HUB;
				while (usKv < pstMap->m_CHub.MrSetGetCount()) {
					if (true == g_bDebug) {
						MR_PRINT(" MR_MR_HUB[%d],", ((MRJBHOST_S*)pstMap->m_CHub.MrSetIndexGet(usKv))->m_uiID);
					}
					stOut.m_iLength = strlen(((MRJBHOST_S*)pstMap->m_CHub.MrSetIndexGet(usKv))->m_unAddr.m_szUrl);
					stOut.m_puszValue = (UCHAR*)&((MRJBHOST_S*)pstMap->m_CHub.MrSetIndexGet(usKv))->m_unAddr.m_szUrl;
					MR_PR_CURMLL += MrProKvOutput(&stOut, MR_PRO_1, MR_PR_CURMLP + MR_PR_CURMLL, MR_BIGPACKET - MR_PR_CURMLL);
					MR_PR_TOTALPP;
					usKv += 1;
				}
			}
			else {
				if (true == g_bDebug) {
					MR_PRINT(" Failed to HubSort\n");
				}
				iReturn = MR_FAILURE;
				break;
			}

			/* glb */
			if (pstMap->MR_PR_MRLEN > 0) {
				if (true == g_bDebug) {
					MR_PRINT(" MR_MR_MR,");
				}
				MR_PR_TOTALPP;
				stOut.m_ucKey = MR_MR_MR;
				stOut.m_iLength = pstMap->MR_PR_MRLEN;
				stOut.m_puszValue = pstMap->MR_PR_MRPKT;
				MR_PR_CURMLL += MrProKvOutput(&stOut, MR_PRO_1, MR_PR_CURMLP + MR_PR_CURMLL, MR_BIGPACKET - MR_PR_CURMLL);
			}

			/* lcl */
			if (true == g_bDebug) {
				MR_PRINT(" MR_MR_LCL\n");
			}
			MR_PR_TOTALPP;
			MR_PR_CURMLL += MrProKvOutput(pstLcl[ucMap], MR_PRO_1, MR_PR_CURMLP + MR_PR_CURMLL, MR_BIGPACKET - MR_PR_CURMLL);

			/* crt */
			CPro.MrProPutBody(NULL);
			CPro.MrProPutAttr(MR_MR_MAP);
			CPro.MrProPutResponse(MR_PRO_Y);
			CPro.MrProPutResult(MR_DEF_SUCC);
			CPro.MrProPutLength(MR_PR_CURMLL - MR_PRO_HEAD);
			CPro.MrProCreate(MR_PR_CURMLP, MR_BIGPACKET);

			/* attr */
			MR_PR_CURMAP.m_ucID = ucMap;
			MR_PR_CURMAP.m_ucStatus |= MR_PR_CRTE;
			if (MR_PR_CURMAP.m_uiHost[MR_PRO_1] > 0) {
				memset(&stPacket, '\0', sizeof(MRJBPACKET_S));
				MrJbDevReqAry(&stPacket, MR_PR_CURMAP.m_uiHost);
				memset(MR_PR_CURMAP.m_uiHost, '\0', sizeof(UINT) * MR_PR_HOST);
				if (MR_DEF_SUCC == MrJbDevSort(CUdp, &stPacket, MR_PR_CURMAP.m_uiHost)) {
					if (true == g_bDebug) {
						MR_DEBUG("Success to Devsort map[%d] -> ", ucMap);
						MrJbDevDisplay(MR_PR_CURMAP.m_uiHost);
					}
				}
				else {
					MR_ERROR("Failure to DevSort map[%d]\n", ucMap);
					iReturn = MR_FAILURE;
					break;
				}
			}

			ucMap += 1;
		}
		CUdp.MrNetClose();
	}
	else {
		iReturn = MR_FAILURE;
	}

	return iReturn;
}

int MrProcessRedCreate(MRPRJOB_S* pstMap, MRPRJOB_S* pstRed)
{
	CMrPro CPro;
	UCHAR ucMap = 0;
	UCHAR ucRed = 0;
	USHORT usKv = 0;
	MRPROKV_S stOut;
	MRPROKV_S stIn[MR_JB_PAIR];

	/* init */
	pstRed->m_pszAdp = pstMap->m_pszAdp;
	pstRed->m_CHub.MrSetClone(pstMap->m_CHub);
	pstRed->m_ucTotal = pstRed->m_CHub.MrSetGetCount();
	memcpy(&pstRed->m_stGlobal, &pstMap->m_stGlobal, sizeof(MRJBPACKET_S));

	/* glb */
	if (true == g_bDebug) {
		MR_DEBUG("RedCreate:");
	}
	ucRed = 0;
	stOut.m_ucKey = MR_MR_MR;
	stOut.m_iLength = pstRed->MR_PR_MRLEN;
	while (ucRed < pstRed->m_CHub.MrSetGetCount()) {
		if (true == g_bDebug) {
			MR_PRINT(" Red[%d].MR_MR_MR,", ucRed);
		}
		MR_PR_CURRLL = MR_PRO_HEAD;
		stOut.m_puszValue = pstRed->MR_PR_MRPKT;
		MR_PR_CURRLL += MrProKvOutput(&stOut, MR_PRO_1, MR_PR_CURRLP + MR_PR_CURRLL, MR_BIGPACKET - MR_PR_CURRLL);
		MR_PR_CURRED.m_ucCur += 1;
		ucRed += 1;
	}

	/* lcl */
	while (ucMap < pstMap->m_ucTotal) {
		CPro.MrProParse(MR_PR_CURMAP.MR_PR_LCLPKT, MR_PR_CURMAP.MR_PR_LCLLEN);
		if (CPro.MrProGetLength() > 0) {
			usKv = 0;
			ucRed = 0;
			MrReset(stIn, sizeof(MRPROKV_S) * MR_JB_PAIR);
			MrProKvInput(stIn, CPro.MrProGetTotal(), CPro.MrProGetBody(), CPro.MrProGetLength());
			while (usKv < CPro.MrProGetTotal()) {
				if (true == g_bDebug) {
					MR_PRINT(" Red[%d].MR_MR_LCL,", ucRed);
				}
				MR_PR_CURRLL += MrProKvOutput(&stIn[usKv], MR_PRO_1, MR_PR_CURRLP + MR_PR_CURRLL, MR_BIGPACKET - MR_PR_CURRLL);
				MR_PR_CURRED.m_ucCur += 1;
				ucRed += 1;
				usKv += 1;
			}
		}
		ucMap += 1;
	}
	if (true == g_bDebug) {
		MR_PRINT("\n");
	}

	/* crt */
	ucRed = 0;
	MrProcessRedSort(pstRed);
	while (ucRed < pstRed->m_ucTotal) {
		if (MR_PR_CURRLL > (int)(MR_PRO_HEAD + pstRed->MR_PR_MRLEN)) {
			CPro.MrProPutBody(NULL);
			CPro.MrProPutAttr(MR_MR_RD1);
			CPro.MrProPutResponse(MR_PRO_Y);
			CPro.MrProPutResult(MR_DEF_SUCC);
			CPro.MrProPutTotal(MR_PR_CURRED.m_ucCur);
			CPro.MrProPutLength(MR_PR_CURRLL - MR_PRO_HEAD);
			CPro.MrProCreate(MR_PR_CURRLP, MR_BIGPACKET);

			MR_PR_CURRED.m_ucID = ucRed;
			MR_PR_CURRED.m_ucCur = MR_PRO_0;
			MR_PR_CURRED.m_ucStatus |= MR_PR_CRTE;
		}
		else {
			MR_PR_CURRLL = MR_PRO_0;
		}
		ucRed += 1;
	}

	return MR_SUCCESS;
}

int MrProcessJobSend(MRPRJOB_S* pstJob, CMrSet& CDev)
{
	CMrPro CPro;
	int iLength = 0;
	UCHAR ucJob = 0;
	MRJBHOST_S stHost;
	bool bBreak = false;
	int iReturn = MR_SUCCESS;
	time_t ulTime = time(NULL);
	UCHAR uszHead[MR_PRO_HEAD] = { 0 };

	do {
		if (time(NULL) - ulTime > MR_TIMEOUT) {
			iReturn = MR_FAILURE;
			break;
		}

		/* send */
		for (ucJob = 0; ucJob < pstJob->m_ucTotal; ucJob ++) {
			if (MR_PRO_0 == MR_PR_CURJOB.m_CUdp.MrNetGetSocket()) {
				if (MR_SUCCESS == (iReturn = MR_PR_CURJOB.m_CUdp.MrNetOpen(MR_SOCK_DGRAM))) {
					if (true == g_bDebug) {
						MR_DEBUG("MrNetOpen task[%d]: %d\n", ucJob, MR_PR_CURJOB.m_CUdp.MrNetGetSocket());
					}
					MR_PR_CURJOB.m_CUdp.MrNetSetTimeOut(0, MR_USEC);
				}
				else {
					MR_ERROR("Failed to MrNetOpen task[%d]\n", ucJob);
					break;
				}
			}
			if (MR_PR_RESP != (MR_PR_CURJOB.m_ucStatus & MR_PR_RESP)) {
				while (MR_PR_CURJOB.m_ucCur < MR_PR_HOST) {
					if (MR_PR_CURJOB.m_uiHost[MR_PR_CURJOB.m_ucCur] > 0) {
						if (MR_SUCCESS == CDev.MrSetSearch(&MR_PR_CURJOB.m_uiHost[MR_PR_CURJOB.m_ucCur], &stHost)) {
							if (true == g_bDebug) {
								MR_DEBUG("MrNetSend task[%d].host[%d]\n", ucJob, stHost.m_uiID);
							}
							if (MR_PR_CURJOB.m_CUdp.MrNetGetSocket() > 0) {
								bBreak = true;
								MR_PR_CURJOB.m_ucStatus |= MR_PR_SEND;
								MR_PR_CURJOB.m_CUdp.MrUdpSetAddr(&stHost.m_unAddr.m_stAddr);
								MR_PR_CURJOB.m_CUdp.MrNetSend(MR_PR_CURJOB.MR_PR_LCLPKT, MR_PR_CURJOB.MR_PR_LCLLEN);
							}
						}
						else {
							MR_ERROR("Failed to MrSetSearch %d\n", MR_PR_CURJOB.m_uiHost[MR_PR_CURJOB.m_ucCur]);
						}
					}
					else {
						MR_PR_CURJOB.m_ucStatus |= MR_PR_FAIL;
						iReturn = MR_FAILURE;
						break;
					}
					MR_PR_CURJOB.m_ucCur += 1;
					if (true == bBreak) {
						break;
					}
				}
			}
		}

		/* recv */
		if (MR_SUCCESS == iReturn) {
			for (ucJob = 0; ucJob < pstJob->m_ucTotal; ucJob ++) {
				if (MR_PR_RESP != (MR_PR_CURJOB.m_ucStatus & MR_PR_RESP)) {
					if (MR_PR_CURJOB.m_CUdp.MrNetGetSocket() > 0) {
						if ((iLength = MR_PR_CURJOB.m_CUdp.MrNetRecv(uszHead, MR_PRO_HEAD)) > 0) {
							CPro.MrProParse(uszHead, iLength);
							if (MR_DEF_SUCC == CPro.MrProGetResult()) {
								if (true == g_bDebug) {
									MR_DEBUG("Success to MrNetRecv task[%d] reponse\n", ucJob);
								}
								MR_PR_CURJOB.m_ucStatus |= MR_PR_RESP;
								if (0 == pstJob->MR_PR_MRLEN) {
									MR_PR_CURJOB.m_CUdp.MrNetClose();
								}
							}
							else {
								MR_ERROR("Failed to MrProGetResult task[%d]\n", ucJob);
							}
						}
						else {
							MR_ERROR("Failure to MrNetRecv task[%d] reponse\n", ucJob);
						}
					}
				}
			}
		}
		else {
			break;
		}
	} while (false == MrProcessJobStatus(pstJob, MR_PR_RESP | MR_PR_FAIL));

	if (MR_FAILURE == iReturn) {
		if (true == g_bDebug) {
			MR_DEBUG("Failed to JobSend\n");
		}
		for (ucJob = 0; ucJob < pstJob->m_ucTotal; ucJob ++) {
			MR_PR_CURJOB.m_CUdp.MrNetClose();
		}
	}

	return iReturn;
}

int MrProcessJobRecv(MRPRJOB_S* pstJob)
{
	CMrPro CPro;
	UCHAR ucJob = 0;
	int iReturn = MR_SUCCESS;
	time_t ulTime = time(NULL);

	do {
		if (time(NULL) - ulTime > MR_TIMEOUT) {
			iReturn = MR_FAILURE;
		}

		if (MR_SUCCESS == iReturn) {
			for (ucJob = 0; ucJob < pstJob->m_ucTotal; ucJob ++) {
				if (MR_PR_CURJOB.m_CUdp.MrNetGetSocket() > 0) {
					MR_PR_CURJOB.MR_PR_LCLLEN = MR_PR_CURJOB.m_CUdp.MrNetRecv(MR_PR_CURJOB.MR_PR_LCLPKT, MR_BIGPACKET);
					if (MR_PR_CURJOB.MR_PR_LCLLEN  > 0) {
						if (true == g_bDebug) {
							MR_DEBUG("Recv task[%d] -> length: %d\n", ucJob, MR_PR_CURJOB.MR_PR_LCLLEN);
						}
						MR_PR_CURJOB.m_ucStatus |= MR_PR_RECV;
						MR_PR_CURJOB.m_CUdp.MrNetClose();
					}
					else {
						MR_ERROR("Failed to MrNetRecv task[%d]\n", ucJob);
					}
				}
			}
		}
		else {
			for (ucJob = 0; ucJob < pstJob->m_ucTotal; ucJob ++) {
				MR_PR_CURJOB.m_CUdp.MrNetClose();
			}
			break;
		}
	} while (false == MrProcessJobStatus(pstJob, MR_PR_RECV));

	return iReturn;
}

int MrProcessResCreate(MRPRJOB_S* pstJob, UCHAR* puszResult, int* piResult)
{
	CMrPro CPro;
	MRPROKV_S stKv;
	UCHAR ucJob = 0;
	USHORT usTotal = 0;

	if (true == g_bDebug) {
		MR_DEBUG("ResCreate:");
	}
	/* adp */
	if (true == g_bDebug) {
		MR_PRINT(" MR_MR_ADP");
	}
	usTotal += 1;
	*piResult = MR_PRO_HEAD;
	stKv.m_ucKey = MR_MR_ADP;
	stKv.m_iLength = strlen(pstJob->m_pszAdp);
	stKv.m_puszValue = (UCHAR*)pstJob->m_pszAdp;
	*piResult += MrProKvOutput(&stKv, MR_PRO_1, puszResult + *piResult, MR_BIGPACKET - *piResult);

	/* glb */
	if (true == g_bDebug) {
		MR_PRINT(" MR_MR_MR");
	}
	usTotal += 1;
	stKv.m_ucKey = MR_MR_MR;
	stKv.m_iLength = pstJob->MR_PR_MRLEN;
	stKv.m_puszValue = pstJob->MR_PR_MRPKT;
	*piResult += MrProKvOutput(&stKv, MR_PRO_1, puszResult + *piResult, MR_BIGPACKET - *piResult);

	/* lcl */
	while (ucJob < pstJob->m_ucTotal) {
		CPro.MrProParse(MR_PR_CURJOB.MR_PR_LCLPKT, MR_PR_CURJOB.MR_PR_LCLLEN);
		if (CPro.MrProGetLength() > 0) {
			if (*piResult + CPro.MrProGetLength() < MR_BIGPACKET) {
				if (true == g_bDebug) {
					MR_PRINT(" Red[%d].MR_MR_LCL", ucJob);
				}
				memcpy(puszResult + *piResult, CPro.MrProGetBody(), CPro.MrProGetLength());
				*piResult += CPro.MrProGetLength();
				usTotal += CPro.MrProGetTotal();
			}
			else {
				*piResult = 0;
				break;
			}
		}
		ucJob += 1;
	}
	if (true == g_bDebug) {
		MR_PRINT("\n");
	}

	if (*piResult > 0) {
		CPro.MrProPutBody(NULL);
		CPro.MrProPutTotal(usTotal);
		CPro.MrProPutAttr(MR_MR_RD2);
		CPro.MrProPutResponse(MR_PRO_Y);
		CPro.MrProPutResult(MR_DEF_SUCC);
		CPro.MrProPutLength(*piResult - MR_PRO_HEAD);
		CPro.MrProCreate(puszResult, MR_BIGPACKET);
	}
	else {
		return MR_FAILURE;
	}

	return MR_SUCCESS;
}

int MrProcessResTrack(UCHAR* puszResult, int iResult, UINT* puiRed)
{
	CMrPro CPro;
	CMrUdp CUdp;
	UINT uiCount = 0;
	MRJBHOST_S stHost;
	int iReturn = MR_FAILURE;
	UCHAR uszHead[MR_PRO_HEAD] = { 0 };

	if (MR_SUCCESS == CUdp.MrNetOpen(MR_SOCK_DGRAM)) {
		if (true == g_bDebug) {
			MR_DEBUG("MrNetOpen res: %d\n", CUdp.MrNetGetSocket());
		}
		CUdp.MrNetSetTimeOut(0, MR_USEC);
		while (0 != *(puiRed + uiCount)) {
			if (MR_SUCCESS == g_CRed.MrSetSearch(puiRed + uiCount, &stHost)) {
				if (true == g_bDebug) {
					MR_DEBUG("MrNetSend red[%d] packet\n", *(puiRed + uiCount));
				}
				CUdp.MrUdpSetAddr(&stHost.m_unAddr.m_stAddr);
				if (CUdp.MrNetSend(puszResult, iResult) > 0) {
					if (CUdp.MrNetRecv(uszHead, MR_PRO_HEAD) > 0) {
						CPro.MrProParse(uszHead, MR_PRO_HEAD);
						if (MR_DEF_SUCC == CPro.MrProGetResult()) {
							if (true == g_bDebug) {
								MR_DEBUG("Success to MrNetRecv Red[%d] reponse\n", *(puiRed + uiCount));
							}
							iReturn = MR_SUCCESS;
							break;
						}
					}
					else {
						MR_ERROR("Failed to MrNetRecv\n");
					}
				}
				else {
					MR_ERROR("Failed to MrNetSend\n");
				}
			}
			uiCount += 1;
		}

		CUdp.MrNetClose();
	}

	if (true == g_bDebug) {
		if (MR_SUCCESS == iReturn) {
			MR_DEBUG("MrJob success!!!\n");
		}
		else {
			MR_DEBUG("MrJob failure!!!\n");
		}
	}

	return iReturn;
}

void MrProcessHubSort(CMrUdp& CUdp, CMrSet& CHub)
{
	MRJBHOST_S stHub;
	USHORT usCount = 0;
	UINT uiHub[MR_JB_DEVICE] = { 0 };

	if (MR_SUCCESS == CHub.MrSetInit(g_CHub.MrSetGetCount(), sizeof(MRJBHOST_S))) {
		if (MR_DEF_SUCC == MrJbDevSort(CUdp, &g_stRed, uiHub)) {
			while (0 != uiHub[usCount]) {
				if (MR_SUCCESS == g_CHub.MrSetSearch(&uiHub[usCount], &stHub)) {
					CHub.MrSetIndexPut(CHub.MrSetGetCount(), &stHub);
				}
				usCount += 1;
			}
		}
	}

	return;
}

void MrProcessRedInit(MRPRJOB_S* pstJob, UINT* puiRed)
{
	UCHAR ucJob = 0;
	UCHAR ucTotal = pstJob->m_CHub.MrSetGetCount() < MR_PR_HOST ? pstJob->m_CHub.MrSetGetCount() : MR_PR_HOST;

	while (ucJob < ucTotal) {
		if (0 == (*(puiRed + ucJob) = ((MRJBHOST_S*)pstJob->m_CHub.MrSetIndexGet(ucJob))->m_uiID)) {
			break;
		}
		ucJob += 1;
	}

	if (true == g_bDebug) {
		MR_DEBUG("RedOrder:");
		for (ucJob = 0; ucJob < MR_PR_HOST; ucJob ++) {
			if (*(puiRed + ucJob) > 0) {
				MR_PRINT(" %d,", *(puiRed + ucJob));
			}
			else {
				break;
			}
		}
		MR_PRINT("\n");
	}

	return;
}

void MrProcessRedSort(MRPRJOB_S* pstRed)
{
	UCHAR ucRed = 0;
	UCHAR ucHost = 0;
	UCHAR ucCount = 0;

	while (ucRed < pstRed->m_ucTotal) {
		ucHost = 0;
		ucCount = ucRed;
		if (true == g_bDebug) {
			MR_DEBUG("RedSort ->");
		}
		do {
			if (ucHost < MR_PR_HOST) {
				if (ucCount < pstRed->m_CHub.MrSetGetCount()) {
					MR_PR_CURRED.m_uiHost[ucHost ++] = ((MRJBHOST_S*)pstRed->m_CHub.MrSetIndexGet(ucCount ++))->m_uiID;
					if (true == g_bDebug) {
						MR_PRINT(" Red[%d]: %d,", ucRed, MR_PR_CURRED.m_uiHost[ucHost - 1]);
					}
				}
				else {
					ucCount = 0;
					ucHost -= 0;
				}
			}
			else {
				break;
			}
		} while (ucCount != ucRed);
		if (true == g_bDebug) {
			MR_PRINT("\n");
		}
		ucRed += 1;
	}

	return;
}

int MrProcessJobCheck(MRPRJOB_S* pstJob)
{
	UCHAR ucJob = 0;
	int iReturn = MR_FAILURE;

	while (ucJob < pstJob->m_ucTotal) {
		if (MR_PR_CURJOB.MR_PR_LCLLEN > MR_PRO_HEAD) {
			iReturn = MR_SUCCESS;
			break;
		}
		ucJob += 1;
	}

	return iReturn;
}

bool MrProcessJobStatus(MRPRJOB_S* pstJob, UCHAR ucStatus)
{
	UCHAR ucJob = 0;
	bool bReturn = true;

	while (ucJob < pstJob->m_ucTotal) {
		if (0 == (MR_PR_CURJOB.m_ucStatus & ucStatus)) {
			bReturn = false;
			break;
		}
		ucJob += 1;
	}

	return bReturn;
}

void MrProcessError(char* pszAdp, CMrPro& CPro)
{
	CMrUdp CUdp;
	UCHAR uszError[MR_PRO_HEAD] = { 0 };

	if (MR_SUCCESS == CUdp.MrNetOpen(MR_SOCK_DGRAM)) {
		/* error */
		CPro.MrProPutTotal(MR_PRO_0);
		CPro.MrProPutLength(MR_PRO_0);
		CPro.MrProPutResponse(MR_PRO_N);
		CPro.MrProCreate(uszError, MR_PRO_HEAD);
		/* send */
		CUdp.MrUdpSetAddr(pszAdp);
		CUdp.MrNetSend(uszError, MR_PRO_HEAD);

		CUdp.MrNetClose();
	}

	return;
}
